package org.example

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.functions.desc
import org.apache.spark.sql.functions.col


object sparkSQL_Yun1_1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("sparkBase")
      .getOrCreate()
    val sc = spark.sparkContext
    //表结构读取电影用户数据
    val schemaUser = StructType(Seq(
      StructField("id",IntegerType),
      StructField("gender",StringType),
      StructField("age",IntegerType),
      StructField("occupation",IntegerType),
      StructField("location",StringType)
    ))
    val user = spark.read.option("sep","::").schema(schemaUser).csv("src/main/resources/users.dat")
    user.show(5)
//1、查询where/filter  替换udf
    user.where("gender = 'F' and age = '18'").show(5)
    user.filter("gender = 'F' and age = '18'").show(5)
    spark.udf.register("replace",(x:String) =>{
      x match {
        case "M" => 0
        case "F" => 1
      }
    })
    user.selectExpr("id","replace(gender) as sexual", "age").show(3)
    user.select("id","gender").show(3  )
//2、排序
    user.orderBy(desc("id")).show(5)
    user.sort(-user("id")).show(5)
    //3、分组groupBy
    user.groupBy("gender").count().show()
    //4、连接 join(表，“列名”，“左（右）连接left——outer”)根据列名连接两个表
    //5、将users用户表跟rating评分表还有movies电影表连接起来，查找年龄18岁的女生评分为5分的所有电影

    // 1. 读取数据
    val users = spark.read
      .option("delimiter", "::")
      .schema("UserID INT, Gender STRING, Age INT, Occupation INT, ZipCode STRING")
      .csv("src/main/resources/users.dat")

    val ratings = spark.read
      .option("delimiter", "::")
      .schema("UserID INT, MovieID INT, Rating INT, Timestamp LONG")
      .csv("src/main/resources/ratings.dat")

    val movies = spark.read
      .option("delimiter", "::")
      .schema("MovieID INT, Title STRING, Genres STRING")
      .csv("src/main/resources/movies.dat")

    // 2. 表关联
    val user_ratings = users.join(ratings, "UserID")
    val full_data = user_ratings.join(movies, "MovieID")

    // 3. 筛选与去重
    val result = full_data
      .filter(col("Age") === 18 && col("Gender") === "F" && col("Rating") === 5)
      .select("T mnitle")
      .distinct()

    result.show(truncate = false)


    sc.stop()
  }
}
